package csl.mq.rabbitmq.old.agent.config;

import cn.hutool.json.JSONObject;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import com.rabbitmq.client.ShutdownSignalException;
import csl.mq.rabbitmq.old.agent.AsyncEventRes;
import csl.mq.rabbitmq.old.agent.EventConsumer;
import csl.mq.rabbitmq.old.agent.EventProcess;
import csl.mq.rabbitmq.old.agent.MessageSender;
import csl.mq.rabbitmq.old.base.BaseMessage;
import csl.mq.rabbitmq.utils.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.ConsumerCancelledException;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.MessageConverter;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

@Slf4j
public class AsyncEventAccessBuilder<T extends BaseMessage> {

    private final ConnectionFactory connectionFactory = SpringUtils.getBean("connectionFactory");

//    private final RedisUtils redisUtils = SpringUtils.getBean("redisUtils");

    public MessageSender<T> buildTopicMessageSender(final String exchange, final String deadExChange, final String routingKey) throws IOException {
        return buildMessageSender(exchange, deadExChange, routingKey, null, "topic");
    }

    public MessageSender<T> buildMessageSender(final String exchange, final String deadExChange, final String routingKey, final String queue, final String type) throws IOException {
        Connection connection;
        if (type.equals("direct")) {
            connection = connectionFactory.createConnection();
            buildQueue(exchange, deadExChange, routingKey, queue, connection, "direct");
        } else if (type.equals("topic")) {
            connection = connectionFactory.createConnection();
            buildTopic(exchange, deadExChange, connection);
        }
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setExchange(exchange);
        rabbitTemplate.setRoutingKey(routingKey);
        rabbitTemplate.setMessageConverter(new JsonMessageConverter());
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.info("send message failed: " + cause + correlationData.toString());
            }
        });
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("send message failed: " + replyCode + " " + replyText);
            rabbitTemplate.send(message);
        });
        return message -> {
            try {
                rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(String.valueOf(message.getMsgId())));
            } catch (Exception e) {
                return new AsyncEventRes(false, e.getMessage());
            }
            return new AsyncEventRes(true, "");
        };
    }

    public EventConsumer buildMessageConsumer(String exchange, String deadExchange, String routingKey, final String queue, final EventProcess messageProcess) throws IOException {
        return buildMessageConsumer(exchange, deadExchange, routingKey, queue, messageProcess, "direct");
    }

    public EventConsumer buildTopicMessageConsumer(String exchange, String deadExchange, String routingKey, final String queue, final EventProcess messageProcess) throws IOException {
        return buildMessageConsumer(exchange, deadExchange, routingKey, queue, messageProcess, "topic");
    }

    public EventConsumer buildMessageConsumer(String exchange, String deadExChange, String routingKey, final String queue, final EventProcess messageProcess, String type) throws IOException {
        Connection connection;
        if (type.equals("direct")) {
            connection = connectionFactory.createConnection();
            buildQueue(exchange, deadExChange, routingKey, queue, connection, "direct");
        } else if (type.equals("topic")) {
            connection = connectionFactory.createConnection();
            buildTopic(exchange, deadExChange, connection);
        } else {
            connection = connectionFactory.createConnection();
        }
        buildQueue(exchange, deadExChange, routingKey, queue, connection, type);
        final MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
        final MessageConverter messageConverter = new JsonMessageConverter();
        return new EventConsumer() {
            Channel channel;

            {
                channel = connection.createChannel(false);
            }

            @Override
            public AsyncEventRes consume() {
                try {
                    GetResponse response = channel.basicGet(queue, false);
                    while (response == null) {
                        response = channel.basicGet(queue, false);
                        Thread.sleep(1000);
                    }
                    Message message = new Message(response.getBody(), messagePropertiesConverter.toMessageProperties(response.getProps(), response.getEnvelope(), "UTF-8"));
                    @SuppressWarnings("unchecked")
                    JSONObject messageBean = (JSONObject) messageConverter.fromMessage(message);
                    AsyncEventRes asyncEventRes = null;
                    String cacheKey = "ASYNC_EVENT:" + message.getMessageProperties().getReceivedRoutingKey();
                    try {
//                        long incrNum = redisUtils.incrLong(cacheKey, 30, TimeUnit.SECONDS);
//                        if (incrNum != 0L) {
//                            asyncEventRes = new AsyncEventRes(false, "Duplicate async event: " + cacheKey);
//                        } else {
//                            asyncEventRes = messageProcess.process(messageBean);
//                        }
                    } catch (Exception e) {
                        log.error("exception", e);
                        asyncEventRes = new AsyncEventRes(false, "process exception: " + e);
                    } finally {
//                        redisUtils.deleteObject(cacheKey);
                    }
                    if (asyncEventRes.isSuccess()) {
                        channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
                    } else {
                        Thread.sleep(1000);
                        log.info("process message failed: " + asyncEventRes.getErrMsg() + ".Route Key:" + message.getMessageProperties().getReceivedRoutingKey());
                        channel.basicNack(response.getEnvelope().getDeliveryTag(), false, true);
                    }

                    return asyncEventRes;
                } catch (InterruptedException e) {
                    log.error("exception", e);
                    return new AsyncEventRes(false, "interrupted exception " + e);
                } catch (ShutdownSignalException | ConsumerCancelledException | IOException e) {
                    log.error("exception", e);

                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        log.error("exception", ex);
                    }

                    channel = connection.createChannel(false);

                    return new AsyncEventRes(false, "shutdown or cancelled exception " + e);
                } catch (Exception e) {
                    log.info("exception : ", e);

                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        ex.printStackTrace();
                    }

                    channel = connection.createChannel(false);

                    return new AsyncEventRes(false, "exception " + e);
                }
            }
        };
    }

    private void buildQueue(String exchange, String deadExChange, String routingKey, final String queue, Connection connection, String type) throws IOException {
        Channel channel = connection.createChannel(false);

        if (type.equals("direct")) {
            channel.exchangeDeclare(exchange, "direct", true, false, null);
        } else if (type.equals("topic")) {
            channel.exchangeDeclare(exchange, "topic", true, false, null);
        }

        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000);
        args.put("x-dead-letter-exchange", deadExChange);
        args.put("x-dead-letter-routing-key", routingKey + "_dead");
        channel.queueDeclare(queue, true, false, false, args);
        channel.queueBind(queue, exchange, routingKey);

        //声明死信队列
        channel.exchangeDeclare(deadExChange, type, true, false, null);
        channel.queueDeclare(queue + "_dead", true, false, false, null);
        channel.queueBind(queue + "_dead", deadExChange, routingKey + "_dead");


        log.info("exchange:{}, deadExChange:{}, routingKey:{}, queue:{}, type:{}", exchange, deadExChange, routingKey, queue, type);
        try {
            channel.close();
        } catch (TimeoutException e) {
            log.info("close channel time out ", e);
        }
    }

    private void buildTopic(String exchange, String deadExChange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", deadExChange);
        channel.exchangeDeclare(exchange, "topic", true, false, args);
    }

}
